Kubeflow Pipelines

ここでは Kubeflow Pipelines 一般についてまとめる

Notebookx や Profile や Dashboard や Serving 的なものはしらない、パイプラインのとこだけ使ってる

事前情報

DSL がしんどくて特におすすめはしない

Python コードとしては valid でも実際に動かそうとするとうまくいかなくてハマることが多い

Type Hint を入出力の型に使っていて、型は合っていてもジェネリクスがダメとか実行時は違うとかでハマる

Python を実行しているつもりでいてはいけない、Python コードによって生成されたパイプラインが実行されている

コンポーネントごとに実行時に GPU 付けて終わったら捨てる等ができる、サーバーレスに近い体験、そして安い

世界観

リモートで動かすものである、ローカルのファイル渡したり読んだりはできない

単純な値なら pipeline 引数に渡す / そうでないなら Cloud Storage に置いて参照する

事前に DAG (有効非巡回グラフ) を構築して動かすものである

動的にフローが変わったり、実行するステップ数が変わるようなものは向いてないorできない

ランタイムは実質 Vertex AI Pipeline 用? OSS のバックエンドで動かない機能がちりばめられててそれでいいのか

コンポーネントとパイプライン

パイプラインはコンポーネントを呼び出してつないでいく

パイプライン自体をコンポーネントとして別のパイプラインから呼べる

値を返す時もコンポーネントと同じ

コンポーネントは3種

1関数に閉じて(Hermetic に)実装(import は全部関数内で書く)

依存関係の固定にはこちらがよいが、イメージ更新 & 実行 の 2 ステップが要るので小さいうちはやらないかな

生成される Dockerfile を書き換えて運用したりしてるの?

これ使わず別のビルドパイプラインで Docker イメージを作って、Lightweight か Container にしがち

任意のコンテナを実行する、実装内容は自由になる

コマンド引数に渡せるものしか入出力できない、簡単な入出力以外は全てファイルパス渡す形になる

機能が限定されすぎている Artifact やら Metadata 書くには後続にもう1コンポーネント作る羽目になる

書いたコードがそのまま実行されてるわけじゃない、これは DSL だ!!!

何度繰り返してもいい概念

DSL メモ

コンポーネント入出力の型は常に書く、呼ぶ時も常に名前付き引数で呼ぶ、実行時に参照される

型

kfp.dsl.base_component.PipelineTask

コンポーネントへの入出力

コンポーネントの Type Hint で入出力を表現している、型チェック時だけのおまけ扱いじゃない

DSL として使えるのは

基本的な型

いや型パラメータ読み飛ばしてくれよ...と思うけどダメ、知らないと普通に型付けてめちゃハマる

output, input は型パラメータのものに変換可能

Output の2つの書き方

コンポーネント引数で Output を宣言する (Traditional artifact syntax)

引数に受け取ったオブジェクトに値をセットする、オブジェクトの持つパスに内容を書き込む

code:input_output.py

@dsl.component

with open(dataset.path) as f:

contents = f.read()

# ... train tf_model model on contents of dataset ...

tf_model.save(model.path)

tf_model.metadata = { ... }

関数の返り値型に書く & 値を返す (Pythonic artifact syntax)

Container Component には対応してない(Python コードをそのまま実行しないので)

code:output.py

@dls.component

def my_component(source_uri: str) -> dsl.Dataset:

...

out = dsl.Dataset(

uri=dsl.get_uri(suffix="foo.json"),

metadata={"model": model_name, "source_uri": source_uri},

)

with open(out.path, 'w') as f:

f.write(...)

return out

DSL 記述上に型書いてるだけで、実行時は違う

それ以外は Channel や Placeholder が渡ってくる

イメージ図

url をコンテナで fetch して Dataset にして返すパイプライン

https://gyazo.com/21d583cae9bffc1e881f5d1dd5914558

複数の Output

Hermetic に書く都合上、関数のシグネチャと値作るところで2回書くことになる

https://gyazo.com/af35e5e39141423083833518a4324b0c

代替案

code:namedtupleoutput.py

@dsl.component

def my_comp1() -> NamedTuple("Outputs", a=int, b=str): # type: ignore

from collections import namedtuple

return Outputs(1, "hello")

# あるいは

@dsl.component

def my_comp2() -> NamedTuple("Outputs", a=int, b=str): # type: ignore

return Outputs(1, "hello")

Output & OutputPath は Input, Artifact に変換可能

code:output.py

@dsl.component

with open(output.path, "w") as f:

f.write("traditional")

@dsl.component

def pythonic_output() -> dsl.Artifact:

output = dsl.Artifact(uri=dsl.get_uri())

with open(output.path, "w") as f:

f.write("pythonic")

@dsl.component

def output_path(output: dsl.OutputPath(dsl.Artifact)): # type: ignore

with open(output, "w") as f:

f.write("output path")

@dsl.component

def count_file(input: dsl.Artifact) -> int:

with open(input.path, "r") as f:

return len("".join(f.readlines()))

@dsl.pipeline

def pipeline():

pythonic_output_task = pythonic_output()

count_file(input=pythonic_output_task.output)

traditional_output_task = traditional_output()

count_file(input=traditional_output_task.output)

output_path_task = output_path()

count_file(input=output_path_task.output)

条件分岐

code:quote_if.py

@dsl.pipeline

def my_pipeline():

coin_flip_task = flip_three_sided_coin()

with dsl.If(coin_flip_task.output == 'heads'):

print_comp(text='Got heads!')

with dsl.Elif(coin_flip_task.output == 'tails'):

print_comp(text='Got tails!')

with dsl.Else():

print_comp(text='Draw!')

TODO 両辺で output いけるかみる

ループ

パイプラインとしてはつながらないが複数のコンポーネントを呼ぶことはできる

sleep したりしてもコンパイル遅いだけで意味ないです

dsl.ParallelFor

並列度指定できる

code:quote_parallel.py

@dsl.pipeline

def my_pipeline():

with dsl.ParallelFor(

parallelism=2

) as epochs:

train_model(epochs=epochs)

code:example.py

with dsl.ParallelFor(

parallelism=1

) as item:

task1 = my_component(..., number=item.a)

task2 = my_component(..., number=item.b)

pipeline parameters を使う方法についてコメントしておいた

なのでフラットなオブジェクトを PararellFor に渡す(大抵呼び出す側の Pipeline にも)必要がある

code:name.py

with dsl.ParallelFor(items=items) as item:

task1 = my_component(..., name=item.name) # name="items-loop-item" になってる

LoopParameterArgument は PipelineParameterChannel を継承していて name 持っている

最悪

中の実装見て使ってない名前を選ぶ必要があり、発狂する

https://gyazo.com/c4af35880372ea58e4c39047a92e8b1d

壁殴る代わりに Issue 立てたい

Lightweight Python Component で import する

Container Component の出力

ディレクトリ自分で作らないといけない、いやそれは作ってよ

code:container_component.py

@dsl.container_component

def curl(url: str, content: dsl.OutputPath(str)): # type: ignore

return dsl.ContainerSpec(

image="curlimages/curl:8.6.0",

)

コンポーネントへの入力型を信用するな

一見型チェック通ってても関係ない諦めろ

kfp.dsl.pipeline_channel.PipelineParameterChannel

kfp.dsl.placeholders.InputValuePlaceholder

code:input_types.py

@dsl.container_component

def fetch_url(url: str, content: dsl.OutputPath(str)): # type: ignore

print(type(url)) # kfp.dsl.placeholders.InputValuePlaceholder

return dsl.ContainerSpec(

image="curlimages/curl:8.6.0",

)

@dsl.component

def save_as_artifact(content: str, metadata: dict = {}) -> dsl.Dataset:

print(type(content)) # これは str

out = dsl.Dataset(uri=dsl.get_uri(suffix="content.html"), metadata=metadata)

with open(out.path, "w") as f:

f.write(content)

return out

@dsl.pipeline

def download_pipeline(url: str) -> dsl.Dataset:

print(type(url)) # kfp.dsl.pipeline_channel.PipelineParameterChannel

fetch_task = fetch_url(url=url)

return save_as_artifact(

# metadata={"url": url},

# こう渡せば行けそうに見えるが、PipelineParameterChannel の値を待って解決する人がいないのでダメ

).output

@dsl.pipeline

def pipeline() -> None:

https://gyazo.com/21d583cae9bffc1e881f5d1dd5914558

Importer

外部の Artifact を取り込むやつ、取り込むことで ML Metadata に記録される

実行環境でアクセスできる URI を渡す形だが、Metadata の記録が不要なら単に URI 文字列を渡せばよいので、

モデルへの入力を残したいモチベーションがある時に使う

ローカル実行

code:local.py

from kfp import dsl, local

local.init(runner=local.DockerRunner())

...

if __name__ == "__main__":

pipeline() # 単に実装した pipeline を呼び出す

動作確認やコンパイルの確認に便利だが、色々対応してないのでまあコンパイル通るだけよりちょっとマシなぐらい

dsl.Collected is not yet supported by the KFP orchestration backend, but may be supported by other orchestration backends.

other orchestration backends て何? Vertex AI Pipeline 以外に使えるのあんの?

メタデータ変数

dsl.PIPELINE_JOB_NAME_PLACEHOLDER

パイプライン名-yyyyMMddHHmmss

dsl.PIPELINE_JOB_RESOURCE_NAME_PLACEHOLDER

projects/{project_number}/locations/{location}/pipeineJobs/PIPELINE_JOB_NAME

dsl.PIPELINE_JOB_ID_PLACEHOLDER

数値の ID

dsl.PIPELINE_TASK_NAME_PLACEHOLDER

実行 Component 名

dsl.PIPELINE_TASK_ID_PLACEHOLDER

コンポーネントID? コンポーネントごとの出力に使われる

dsl.PIPELINE_JOB_CREATE_TIME_UTC_PLACEHOLDER

UTC 時刻

dsl.PIPELINE_JOB_SCHEDULE_TIME_UTC_PLACEHOLDER

epoch 0、スケジュール実行なら値が入る

dsl.PIPELINE_ROOT_PLACEHOLDER

pipeline_root

pipeline 内で参照できる、component で見るには渡さないといけない

component で直接参照してもダメ